package com.xrr.nettytest;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class MqttNettyServer {

    private static class MqttServerHandler extends SimpleChannelInboundHandler<MqttMessage>{



        public  Map<String,ChannelHandlerContext> clientMap  =new ConcurrentHashMap<>();

        private Map<String,Set<String>> clientSubscribeTopicMap = new ConcurrentHashMap<>();

        private Map<String,Set<String>> topicSubscribeClientMap = new ConcurrentHashMap<>();


        private Map<String,List<MqttMessage>> remainMessageListMap = new ConcurrentHashMap<>();

        private Map<String,MqttQoS> topicQosMap = new ConcurrentHashMap<>();



        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("客户端:"+ctx.channel().remoteAddress()+"连接上");
            clientMap.put(ctx.channel().id().toString(),ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);
        }

        private void clientConnect(MqttMessage mqttMessage,ChannelHandlerContext channelHandlerContext){
            //	在一个网络连接上，客户端只能发送一次CONNECT报文。服务端必须将客户端发送的第二个CONNECT报文当作协议违规处理并断开客户端的连接
            //	建议connect消息单独处理，用来对客户端进行认证管理等 这里直接返回一个CONNACK消息

            MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;

            MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
            MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();
            //构建返回报文， 可变报头
            MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeaderInfo.isCleanSession());
            //构建返回报文， 固定报头 至多一次(至少—次,只有一次)
            MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK, mqttFixedHeaderInfo.isDup(),MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
            //构建连接回复消息体
            MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
            channelHandlerContext.writeAndFlush(connAck);
        }

        private void  clientSubscribeTopic(MqttMessage mqttMessage,ChannelHandlerContext ctx){

            MqttSubscribeMessage mqttSubscribeMessage  = (MqttSubscribeMessage) mqttMessage;
            MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
            MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
            MqttSubscribePayload subscribePayload = mqttSubscribeMessage.payload();

            String id =ctx.channel().id().toString();

            Set<String> clientSubscribeTopicSet = clientSubscribeTopicMap.get(id);
            if (clientSubscribeTopicSet == null){
                clientSubscribeTopicSet =new HashSet<>();
            }

            //客户端要订阅的主题名和qos级别
            List<MqttTopicSubscription> mqttTopicSubscriptions = subscribePayload.topicSubscriptions();


            List<Integer> grantedQoSLevels = new ArrayList<Integer>();

            Set<String> newSubsribeTopicSet = new HashSet<>();
            for (MqttTopicSubscription subscription : mqttTopicSubscriptions){

                String topicName = subscription.topicFilter();

                //客户端订阅主题map和主题客户端map都要添加
                clientSubscribeTopicSet.add(topicName);
                newSubsribeTopicSet.add(topicName);
               Set<String> topicSubscribeClientSet =  topicSubscribeClientMap.get(topicName);
               topicSubscribeClientSet.add(id);
               grantedQoSLevels.add(subscription.qualityOfService().value());
            }
            //

            //	构建返回报文	有效负载
            MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
            //	构建返回报文	固定报头
            MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
            //	构建返回报文	订阅确认
            MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
            //回写响应
            ctx.writeAndFlush(subAck);
            //todo 判断新订阅的主题是否有消息需要发给新订阅客户端


        }



        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, MqttMessage mqttMessage) throws Exception {
            System.out.println("mqtt消息到达:"+mqttMessage);

            MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();

            switch (mqttFixedHeader.messageType()){

                case CONNECT:
                    clientConnect(mqttMessage,channelHandlerContext);
                    break;
                case SUBSCRIBE:
                    clientSubscribeTopic(mqttMessage,channelHandlerContext);
                    break;
                case UNSUBSCRIBE:
                    clientUnscribeTopic(mqttMessage,channelHandlerContext);
                    break;
                case   PUBLISH:
                    clientPublishQos1Message(mqttMessage,channelHandlerContext);
                    break;
                case PUBREL:
                    clientAckQos2Message(mqttMessage,channelHandlerContext);
                    break;
            }
        }

        private void clientUnscribeTopic(MqttMessage mqttMessage, ChannelHandlerContext channelHandlerContext) {

            MqttUnsubscribeMessage mqttUnsubscribeMessage = (MqttUnsubscribeMessage) mqttMessage;
            List<String> topics = mqttUnsubscribeMessage.payload().topics();

            if (topics!=null && !topics.isEmpty()) {
                Set<String> topicSet = clientSubscribeTopicMap.get(channelHandlerContext.channel().id().toString());
                topics.forEach(topicSet::remove);

                topics.forEach(topic->{
                    topicSubscribeClientMap.get(topic).remove(channelHandlerContext.channel().id().toString());
                });

            }



        }

        private void clientAckQos2Message(MqttMessage mqttMessage, ChannelHandlerContext channelHandlerContext) {


        }

        private void clientPublishQos1Message(MqttMessage mqttMessage, ChannelHandlerContext channelHandlerContext) {

            MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
            //取出topic名称
            MqttPublishVariableHeader mqttPublishVariableHeader = mqttPublishMessage.variableHeader();
            String topicName =mqttPublishVariableHeader.topicName();

            //获取对应主题对应的客户端
            Set<String> channelSet = topicSubscribeClientMap.get(topicName);
            if ( channelSet!=null && !channelSet.isEmpty()){

                for (String channelId : channelSet){

                   ChannelHandlerContext ctx =  clientMap.get(channelId);

                   if (ctx!=null ){

                       if (ctx.channel().isActive()){
                           //发送消息
                           ctx.channel().writeAndFlush(mqttPublishMessage);
                       }else {
                           //移除一已经不活跃的channel
                           clientMap.remove(channelId);
                       }

                   }
                }
            }

        }


    }


    public static void main(String[] args) {



    }

    public static void  start(){

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        ServerBootstrap serverBootstrap = new ServerBootstrap();

        try {
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .option(ChannelOption.SO_REUSEADDR,true)
                    .option(ChannelOption.TCP_NODELAY,true)
                    .option(ChannelOption.SO_KEEPALIVE,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("decoder",new MqttDecoder());
                            pipeline.addLast("encoder", MqttEncoder.INSTANCE);
                            pipeline.addLast(new MqttServerHandler());
                        }
                    });
           ChannelFuture channelFuture  =  serverBootstrap.bind(10021).sync();
           if (channelFuture.isSuccess()){
               System.out.println("mqtt服务端启动");
           }else {
               System.out.println("mqtt服务端启动失败");
           }
        }catch (Exception e){
            e.printStackTrace();
        }
        finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }



}
